package com.ys.mq;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.ys.constant.ShopCode;
import com.ys.entity.MQEntity;
import com.ys.mapper.GoodsMapper;
import com.ys.mapper.GoodsNumberLogMapper;
import com.ys.mapper.MqConsumerLogMapper;
import com.ys.pojo.TradeGoods;
import com.ys.pojo.TradeGoodsNumberLog;
import com.ys.pojo.TradeMqConsumerLog;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.util.Date;

@Slf4j
@Component
@RocketMQMessageListener(topic = "${mq.order.topic}",
        consumerGroup = "${mq.order.consumer.group.name}",
        messageModel = MessageModel.BROADCASTING)
public class GoodsRocketMQLinstener implements RocketMQListener<MessageExt> {

    @Autowired
    private MqConsumerLogMapper mqConsumerLogMapper;

    @Value("${mq.order.consumer.group.name}")
    private String group_name ;

    @Autowired
    private GoodsMapper goodsMapper;

    @Autowired
    private GoodsNumberLogMapper goodsNumberLogMapper;

    @Override
    public void onMessage(MessageExt messageExt) {
        String msgId=null;
        String tags=null;
        String keys=null;
        String body=null;

        //解析消息内容
        msgId = messageExt.getMsgId();
        tags = messageExt.getTags();
        keys = messageExt.getKeys();
        try {
            body  = new String(messageExt.getBody(),"UTF-8");
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }

        log.info("接受消息成功");
        //查询消息消费记录
            TradeMqConsumerLog mqConsumerLog = new TradeMqConsumerLog();
            mqConsumerLog.setMsgTag(tags);
            mqConsumerLog.setMsgKey(keys);
            mqConsumerLog.setMsgBody(body);
            mqConsumerLog.setMsgId(msgId);
        //先查询消息日志里边是否有这条消息
        QueryWrapper<TradeMqConsumerLog> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda().eq(TradeMqConsumerLog::getMsgId,msgId)
                .eq(TradeMqConsumerLog::getMsgBody,body)
                .eq(TradeMqConsumerLog::getMsgKey,keys)
                .eq(TradeMqConsumerLog::getMsgTag,tags);
        TradeMqConsumerLog findlog = mqConsumerLogMapper.selectOne(queryWrapper);
        System.out.println(findlog);

        //查不到消息，就将新消息存入数据库
        if(findlog == null){
            mqConsumerLog.setGroupName(group_name);
            mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_FAIL.getCode());
            mqConsumerLog.setConsumerTimes(0);
            mqConsumerLogMapper.insert(mqConsumerLog);
        }
        //消息存的情况，对消息进行消费
        if(findlog !=null){
            //3. 判断如果消费过...
            //3.1 获得消息处理状态
            Integer status = mqConsumerLog.getConsumerStatus();
            //处理过...返回
            if(ShopCode.SHOP_MQ_MESSAGE_STATUS_SUCCESS.getCode().intValue()==status.intValue()){
                log.info("消息:"+msgId+",已经处理过");
                return;
            }

            //正在处理...返回
            if(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode().intValue()==status.intValue()){
                log.info("消息:"+msgId+",正在处理");
                return;
            }
            //处理失败
            if(ShopCode.SHOP_MQ_MESSAGE_STATUS_FAIL.getCode().intValue()==status.intValue()) {
                //获得消息处理次数
                Integer times = mqConsumerLog.getConsumerTimes();
                if (times > 3) {
                    log.info("消息:" + msgId + ",消息处理超过3次,不能再进行处理了");
                    return;
                }
                mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode());

                //使用数据库乐观锁更新
                //没处理一次times+1
                UpdateWrapper<TradeMqConsumerLog> updateWrapper = new UpdateWrapper<>();
                updateWrapper.lambda().set(TradeMqConsumerLog::getConsumerTimes,mqConsumerLog.getConsumerTimes()+1);
                int update = mqConsumerLogMapper.update(mqConsumerLog, updateWrapper);
                if(update<=0){
                    //未修改成功,其他线程并发修改
                    log.info("并发修改,稍后处理");
                }
            }else {
                //4. 判断如果没有消费过...
                mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode());
                mqConsumerLog.setConsumerTimes(0);
                //将消息处理信息添加到数据库
                mqConsumerLogMapper.insert(mqConsumerLog);
            }

        }
        /*else{
            UpdateWrapper<TradeMqConsumerLog> updateWrapper = new UpdateWrapper<>();
            updateWrapper.lambda().set(TradeMqConsumerLog::getConsumerTimes,mqConsumerLog.getConsumerTimes()+1);
            int update = mqConsumerLogMapper.update(mqConsumerLog, updateWrapper);
        }*/
        //回退库存
        MQEntity mqEntity = JSON.parseObject(body, MQEntity.class);
        Long goodsId = mqEntity.getGoodsId();
        Integer goodsNum = mqEntity.getGoodsNum();
        TradeGoods goods = goodsMapper.selectById(goodsId);
        UpdateWrapper<TradeGoods> updateWrapper = new UpdateWrapper<>();
        updateWrapper.lambda().set(TradeGoods::getGoodsNumber,goodsNum+goods.getGoodsNumber())
                .eq(TradeGoods::getGoodsId,goodsId);
        goodsMapper.update(goods,updateWrapper);
        //记录库存操作日志
        TradeGoodsNumberLog goodsNumberLog = new TradeGoodsNumberLog();
        goodsNumberLog.setOrderId(mqEntity.getOrderId());
        goodsNumberLog.setGoodsId(goodsId);
        goodsNumberLog.setGoodsNumber(mqEntity.getGoodsNum());
        goodsNumberLog.setLogTime(new Date());
        int insert = goodsNumberLogMapper.insert(goodsNumberLog);
        System.out.println("插入goodsNumberlog"+insert);

        //6. 将消息的处理状态改为成功

        UpdateWrapper<TradeMqConsumerLog> logUpdateWrapper = new UpdateWrapper<>();
        logUpdateWrapper.lambda().set(TradeMqConsumerLog::getConsumerStatus,ShopCode.SHOP_MQ_MESSAGE_STATUS_SUCCESS.getCode())
                        .set(TradeMqConsumerLog::getConsumerTimestamp,new Date())
                .eq(TradeMqConsumerLog::getMsgId,msgId)
                .eq(TradeMqConsumerLog::getMsgBody,body)
                .eq(TradeMqConsumerLog::getMsgKey,keys)
                .eq(TradeMqConsumerLog::getMsgTag,tags);
        mqConsumerLogMapper.update(mqConsumerLog,logUpdateWrapper);

        log.info("回退库存成功");

    }
}
